package com.xueyuan.wata.daph.tools

import com.xueyuan.wata.daph.api.config.option.GlobalConfigOptions.{NODE_DICTIONARY_PATH, NODE_JAR_ROOT_PATH}
import com.xueyuan.wata.daph.api.config.{JGlobalConfig, SimpleJGlobalConfig}
import com.xueyuan.wata.daph.api.node.NodeDescription
import org.apache.commons.lang3.StringUtils

import java.io.File
import scala.jdk.CollectionConverters.{collectionAsScalaIterableConverter, seqAsJavaListConverter}

object ConfigTool {
  def simpleConfigToConfig(config: SimpleJGlobalConfig): JGlobalConfig = {
    val res = new JGlobalConfig
    val nodes = config.getNodes.asScala
    val dictionaryPath = config.getOptions.getOrDefault(NODE_DICTIONARY_PATH.key, NODE_DICTIONARY_PATH.default)
    val dictionary = JsonTool.loadFile[Array[NodeInfo]](dictionaryPath)
    val nodeRootPath = config.getOptions.getOrDefault(NODE_JAR_ROOT_PATH.key, NODE_JAR_ROOT_PATH.default)
    val nodePaths = if (nodeRootPath.nonEmpty) {
      val dir = new File(nodeRootPath)
      val fns = dir.listFiles().filter(_.isFile).map(_.getCanonicalPath).toSet
      fns
    } else Set.empty

    res.setOptions(config.getOptions)

    var i = 0
    val resNodes = nodes.map { node =>
      val flag = node.flag
      val nodeInfo = {
        if (flag.contains("-")) {
          dictionary.find(_.alias.contains(flag)).get
        } else {
          dictionary.find(_.flag.equals(flag)).get
        }
      }
      val extraOptions = {
        if (nodePaths.isEmpty) {
          node.extraOptions +
            ("nodePath" -> s"${node.extraOptions("nodePath")}/${nodeInfo.jarName}") +
            ("nodeClassPrefix" -> (
              if (StringUtils.isEmpty(nodeInfo.classPrefix)) nodeInfo.className.split("\\.").dropRight(1).reduce((a, b) => s"$a.$b")
              else nodeInfo.classPrefix))
        } else {
          node.extraOptions +
            ("nodePath" -> nodePaths.find(_.endsWith(nodeInfo.jarName)).get) +
            ("nodeClassPrefix" -> (
              if (StringUtils.isEmpty(nodeInfo.classPrefix)) nodeInfo.className.split("\\.").dropRight(1).reduce((a, b) => s"$a.$b")
              else nodeInfo.classPrefix))
        }
      }

      i += 1
      NodeDescription(
        s"$flag-$i", node.name, nodeInfo.className,
        node.config, node.inLines, node.outLines,
        extraOptions
      )
    }.toList.asJava
    res.setNodes(resNodes)

    res
  }

  private case class NodeInfo(flag: String, alias: Array[String], jarName: String, className: String, classPrefix: String)
}
